package cn.itcast.sql;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author KTL
 * @version V1.0
 * @Package cn.itcast.sql
 * @date 2021/3/6 0006 15:06
 * @Copyright © 2015-04-29  One for each, and two for each
 *     演示flink tablesql 将datastream数据转为table或view 然后使用sql进行统计
 */
public class Demo1 {
    public static void main(String[] args) throws Exception {
        //TODO 0 env
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final EnvironmentSettings build = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        final StreamTableEnvironment tenv = StreamTableEnvironment.create(env, build);
        //TODO 1 source
        final DataStreamSource<Order> orderA = env.fromCollection(Arrays.asList(
                new Order(1L, "beer", 3),
                new Order(1L, "diaper", 4),
                new Order(3L, "rubber", 2)));

        DataStream<Order> orderB = env.fromCollection(Arrays.asList(
                new Order(2L, "pen", 3),
                new Order(2L, "rubber", 3),
                new Order(4L, "beer", 1)));


        //TODO 2 thransformation
        //将DataStream数据转Table和View，然后查询
        Table tableA = tenv.fromDataStream(orderA, $("user"), $("product"), $("amount"));
        tableA.printSchema();
        System.out.println(tableA);
        tenv.createTemporaryView("tableB", orderB,$("user"), $("product"), $("amount"));
        //TODO 3 sink
        String sql ="select * from "+tableA+" where amount >2 \n " +
                "union \n" +
                "select * from tableB where amount>1";
        final Table resultTable = tenv.sqlQuery(sql);
        resultTable.printSchema();
        System.out.println(resultTable);
        final DataStream<Tuple2<Boolean, Order>> resultDS = tenv.toRetractStream(resultTable, Order.class);
        resultDS.print();
        //TODO 0 execute
        env.execute();
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Order{
        public Long user;
        public String product;
        public int amount;
    }
}
